package com.ruyuan.dfs.namenode.datanode;

import com.ruyuan.dfs.common.Constants;
import com.ruyuan.dfs.common.FileInfo;
import com.ruyuan.dfs.common.exception.NameNodeException;
import com.ruyuan.dfs.common.utils.DateUtils;
import com.ruyuan.dfs.common.utils.DefaultScheduler;
import com.ruyuan.dfs.common.utils.PrettyCodes;
import com.ruyuan.dfs.model.datanode.RegisterRequest;
import com.ruyuan.dfs.namenode.config.NameNodeConfig;
import com.ruyuan.dfs.namenode.fs.DiskNameSystem;
import com.ruyuan.dfs.namenode.fs.Node;
import com.ruyuan.dfs.namenode.rebalance.RemoveReplicaTask;
import com.ruyuan.dfs.namenode.rebalance.ReplicaTask;
import com.ruyuan.dfs.namenode.server.UserManager;
import com.ruyuan.dfs.namenode.server.tomcat.domain.User;
import lombok.extern.slf4j.Slf4j;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.stream.Collectors;

/**
 * 负责管理 DataNode
 *
 * @author Sun Dasheng
 */
@Slf4j
public class DataNodeManager {

    //鉴权器
    private final UserManager userManager;
    //<hostName , DataNodeInfo> -> <hostName:port , DataNodeInfo>
    private final Map<String, DataNodeInfo> dataNodes = new ConcurrentHashMap<>();
    //
    private final ReentrantReadWriteLock replicaLock = new ReentrantReadWriteLock();

    /**
     * <pre>
     * 每个文件对应存储的DataNode信息
     *
     * 比如文件 aaa.png，存储在 DataNode01、DataNode02
     *  < fileName ,  dataNodeList >
     *    aaa.png : [
     *        DataNode01,
     *        DataNode02
     *    ]
     * </pre>
     */
    private final Map<String, List<DataNodeInfo>> replicaByFilename = new ConcurrentHashMap<>();

    /**
     * <pre>
     * 每个 DataNode 存储的文件列表
     * < hostName , < fileName , dataNode >>
     * 比如 DataNode01 存储有文件：aaa.jpg、bbb.jpg
     *
     *    DataNode01 : [
     *        aaa.jpg,
     *        bbb.jpg
     *    ]
     * </pre>
     */
    private final Map<String, Map<String, FileInfo>> filesByDataNode = new ConcurrentHashMap<>();

    private final NameNodeConfig nameNodeConfig;
    private DiskNameSystem diskNameSystem;

    public DataNodeManager(NameNodeConfig nameNodeConfig, DefaultScheduler defaultScheduler, UserManager userManager) {
        this.nameNodeConfig = nameNodeConfig;
        this.userManager = userManager;
        long dataNodeAliveThreshold = nameNodeConfig.getDataNodeAliveCheckInterval();
        //todo NameNode 启动  DataNode 存活检测
        defaultScheduler.schedule("DataNode存活检测", new DataNodeAliveMonitor(),
                dataNodeAliveThreshold, dataNodeAliveThreshold, TimeUnit.MILLISECONDS);
    }


    public void setDiskNameSystem(DiskNameSystem diskNameSystem) {
        this.diskNameSystem = diskNameSystem;
    }

    /**
     * dataNode 向 nameNode 进行注册
     * todo 2  DataNode 向 NameNode 进行注册
     *
     * @param request 请求信息
     */
    public boolean register(RegisterRequest request) {
        if (dataNodes.containsKey(request.getHostname())) {
            log.info("dataNode 向 nameNode 进行注册已经存在{}:{}", request.getHostname(), dataNodes);
            return false;
        }
        //构建 NameNode 本地存储 DataNode 信息副本信息
        DataNodeInfo dataNode = new DataNodeInfo(request.getHostname(), request.getNioPort(), request.getHttpPort(),
                System.currentTimeMillis() + nameNodeConfig.getDataNodeHeartbeatTimeout());
        dataNode.setStoredDataSize(request.getStoredDataSize());
        dataNode.setFreeSpace(request.getFreeSpace());
        dataNode.setNodeId(request.getNodeId());
        log.info("收到DataNode注册请求：[hostname={}, storageSize={}, freeSpace={},nodeId={}]",
                request.getHostname(), request.getStoredDataSize(), request.getFreeSpace(), request.getNodeId());
        dataNodes.put(request.getHostname(), dataNode);
        System.out.println(dataNodes);
        log.info("dataNode 向 nameNode 进行注册{}", dataNodes);
        return true;
    }

    /**
     * DataNode 进行心跳
     *
     * @param hostname 主机名
     */
    public Boolean heartbeat(String hostname) {
        DataNodeInfo dataNode = dataNodes.get(hostname);
        if (dataNode == null) {
            return false;
        }
        long latestHeartbeatTime = System.currentTimeMillis() + nameNodeConfig.getDataNodeHeartbeatTimeout();
        if (log.isDebugEnabled()) {
            log.debug("收到DataNode的心跳：[hostname={}, latestHeartbeatTime={}]", hostname, DateUtils.format(new Date(latestHeartbeatTime)));
        }
        dataNode.setLatestHeartbeatTime(latestHeartbeatTime);
        return true;
    }

    /**
     * 获取已经Ready并且排序的DataNode列表
     *
     * @return DataNode列表
     */
    private List<DataNodeInfo> getSortedReadyDataNode() throws NameNodeException {
        final List<DataNodeInfo> dataNodeInfos = dataNodes.values().stream()
                .filter(dataNodeInfo -> dataNodeInfo.getStatus() == DataNodeInfo.STATUS_READY)//心跳时会进行设置的
                .sorted()
                .collect(Collectors.toList());
        if (dataNodeInfos.size() <= 0) {
            throw new NameNodeException("DataNodeManager.getSortedReadyDataNode()：获取已经Ready并且排序的DataNode列表为空 " + dataNodeInfos);
        }
        return dataNodeInfos;
    }

    /**
     * 从DataNode集合中选择节点，但是要排除已经包含该文件的DataNode节点
     * todo NameNodeException: DataNode数量不足: 抛出异常
     *
     * @param dataNodeList      DataNode节点列表
     * @param requiredNodeCount 需要的DataNode节点数量
     * @param filename          文件名
     */
    private List<DataNodeInfo> selectDataNodeFromList(List<DataNodeInfo> dataNodeList, int requiredNodeCount, String filename) throws NameNodeException {
        int existCount = 0;
        long minStoredDataSize = -1;
        /*
         * 在上传小文件的时候，文件大小较小，为了避免流量都打到同一台DataNode机器
         *
         * 所以如在1G多台DataNode之间的存储空间大小误差G范围内，直接随机取一台
         *
         * 如果误差范围小于1G,则按存储空间大小从低到高进行获取
         */
        List<DataNodeInfo> candidateNodes = new ArrayList<>(10);
        //1G
        long delta = 1024 * 1024 * 1024;
        //先遍历所有的节点
        for (DataNodeInfo dataNodeInfo : dataNodeList) {
            //当前节点是否存在这个文件
            if (dataNodeContainsFiles(dataNodeInfo.getHostname(), filename)) {
                existCount++;
                continue;
            }
            //节点的存储情况
            long storedDataSize = dataNodeInfo.getStoredDataSize();
            if (minStoredDataSize < 0) {
                minStoredDataSize = storedDataSize;
            }
            //当前节点从存储情况 - min 小于误差情况
            if (dataNodeInfo.getStoredDataSize() - minStoredDataSize <= delta) {
                candidateNodes.add(dataNodeInfo);
            }
        }

        int findCount = 0;
        //可利用节点 正好正等于 用户设置的 副本数量
        if (candidateNodes.size() == requiredNodeCount) {
            // 误差在1G以内的DataNode数量刚好和需要的节点数量一样
            return candidateNodes;
        } else if (candidateNodes.size() < requiredNodeCount) {
            // 误差在1G以内的 DataNode 数量小于需要的节点数量，则需要从 DataNode 列表中继续取到足够的节点
            int remainNodeCount = requiredNodeCount - candidateNodes.size();
            for (DataNodeInfo dataNodeInfo : dataNodeList) {
                if (candidateNodes.contains(dataNodeInfo) || dataNodeContainsFiles(dataNodeInfo.getHostname(), filename)) {
                    continue;
                }
                candidateNodes.add(dataNodeInfo);
                remainNodeCount--;
                findCount++;
                if (remainNodeCount <= 0) {
                    return candidateNodes;
                }
            }
        } else {
            // 误差在1G以内的DataNode数量很多，超过所需的节点数量，则随机取几个
            Random random = new Random();
            List<DataNodeInfo> selectedDataNodes = new ArrayList<>();
            for (int i = 0; i < requiredNodeCount; i++) {
                int index = random.nextInt(candidateNodes.size());
                DataNodeInfo dataNodeInfo = candidateNodes.get(index);
                if (selectedDataNodes.contains(dataNodeInfo) || dataNodeContainsFiles(dataNodeInfo.getHostname(), filename)) {
                    continue;
                }
                selectedDataNodes.add(dataNodeInfo);
            }
            return selectedDataNodes;

        }
        log.error("DataNode数量不足：[datanodeList={}]", dataNodes.values());
        throw new NameNodeException("DataNode数量不足: [applyCount=" + requiredNodeCount +
                ", findCount=" + findCount +
                ", existsFileNodeCount=" + existCount +
                ", filename=" + filename);
    }

    /**
     * 设置 DataNode 状态为 Ready
     *
     * @param hostname DataNode主机名
     */
    public void setDataNodeReady(String hostname) {
        DataNodeInfo dataNode = dataNodes.get(hostname);
        if (dataNode != null) {
            dataNode.setStatus(DataNodeInfo.STATUS_READY);
        }
    }

    /**
     * 创建丢失副本的复制任务
     *
     * @param dataNodeInfo 宕机的DataNode
     */
    private void createLostReplicaTask(DataNodeInfo dataNodeInfo) {
        // 从内存数据结构中移除DataNode的文件列表并返回移除节点中的存储数据
        Map<String, FileInfo> filesByDataNode = removeFileByDataNode(dataNodeInfo.getHostname());
        if (filesByDataNode == null) {
            return;
        }
        //找到被移除的节点上所保存的数据
        for (FileInfo fileInfo : filesByDataNode.values()) {
            /**
             * 假如一份数据用户指明需要保存3份 ：DataNode_1 DataNode_2 DataNode_3  DataNode_4(无) DataNode_5(无)
             * 那么当 DataNode_1 宕机了 需要以下流程：
             * 1. 找到 DataNode_1 节点所保存的文件集合的映射表
             * 2. 依次根据 DataNode_1 所保存的文件名称 找到其他节点也保存相同数据的节点 (这里DataNode_2 DataNode_3 可随机法选择一个)
             * 3. 然后有了源节点 DataNode_2 ，那么此时 就从 DataNode 全量集合中 依次判断是否存在这个文件 存在则跳过 不存在则进入候选集合
             * 4. 判断用户传入的副本数量 和 候选集合大小 来进行取舍
             * 5. 最后逐一进行创建副本任务
             * 6. 那么哪里会执行这些任务呢？ 答案：NameNode处理DataNode心跳请求时，响应的就是副本任务
             */
            // 依次根据文件名字来 找到其文件所在保存的其他节点之一
            // 当前节点是为了 让其他节点来复制的
            DataNodeInfo sourceDataNode = chooseReadableDataNodeByFileName(fileInfo.getFileName(), dataNodeInfo);
            if (sourceDataNode == null) {
                log.warn("警告：找不到适合的DataNode用来获取文件：" + fileInfo.getFileName());
                continue;
            }
            //为 复制任务 申请副本，申请的 dataNode 需要排除目标 DataNode 的
            // DataNodeInfo destDataNode = (DataNodeInfo) allocateReplicateDataNodes(fileInfo, sourceDataNode);
            List<DataNodeInfo> dataNodeInfos = allocateReplicateDataNodes(fileInfo, sourceDataNode);
            if (dataNodeInfos == null) {
                log.warn("警告：找不到适合的DataNode用来 Rebalance");
                continue;
            }
            // 需要备份当前文件的节点列表
            // 逐一执行 副本复制任务
            for (DataNodeInfo destDataNode : dataNodeInfos) {
                ReplicaTask task = new ReplicaTask(fileInfo.getFileName(), sourceDataNode.getHostname(), sourceDataNode.getNioPort());
                log.info("创建副本复制任务：[filename={}, from={}, to={}]", fileInfo.getFileName(), sourceDataNode.getHostname(), destDataNode.getHostname());
                destDataNode.addReplicaTask(task);
            }
        }
    }

    /**
     * 为 "复制任务" 筛选副本，申请的 DataNode 需要排除目标 DataNode 的
     *
     * @param fileInfo        文件信息
     * @param excludeDataNode 排除的DataNode（源DataNode）
     */
    private List<DataNodeInfo> allocateReplicateDataNodes(FileInfo fileInfo, DataNodeInfo excludeDataNode) {
        //先从全量的 DataNode 中 排除掉 已宕机的节点 && 将要作为 源数据的节点
        //此时 dataNodes 中已经删除掉了 已经宕机的节点了
        List<DataNodeInfo> dataNodeInfos = dataNodes.values().stream()
                .filter(dataNodeInfo -> !dataNodeInfo.equals(excludeDataNode) && dataNodeInfo.getStatus() == DataNodeInfo.STATUS_READY)
                .sorted()
                .collect(Collectors.toList());
        try {
            //从 DataNode 集合中选择出需要进行复制已宕机的 DataNode的数据  的节点，已经排除掉已经包含该文件的 DataNode 节点
            //是因为这里只默认传了一个 1  吗
            List<DataNodeInfo> dataNodesList = selectDataNodeFromList(dataNodeInfos, fileInfo.getReplicaNum(), fileInfo.getFileName());
            //todo 不懂为何只要一个
            // return dataNodesList.get(0);
            //更改过后的
            return dataNodesList;
        } catch (Exception e) {
            log.warn("allocateReplicateDataNodes select node failed.", e);
            return null;
        }
    }

    /**
     * @param username 该文件的所属的用户
     * @param count    申请机器数量
     *                 为文件分配dataNode机器列表
     */
    public List<DataNodeInfo> allocateDataNodes(String username, int count, String filename) throws Exception {
        User user = userManager.getUser(username);
        Set<String> dataNodeSet = user.getStorageInfo().getDataNodesSet();
        if (dataNodeSet.isEmpty()) {
            //获取已经Ready并且排序的DataNode列表
            List<DataNodeInfo> sortedReadyDataNode = getSortedReadyDataNode();
            // 从DataNode集合中选择节点，但是要排除已经包含该文件的DataNode节点
            return selectDataNodeFromList(sortedReadyDataNode, count, filename);
        } else {
            // 这里用户是指定了DataNode的，则在指定的DataNode中查找对应的DataNode
            List<DataNodeInfo> dataNodeInfos = dataNodes.values().stream()
                    .filter(dataNodeInfo -> dataNodeInfo.getStatus() == DataNodeInfo.STATUS_READY
                            && dataNodeSet.contains(dataNodeInfo.getHostname()))
                    .sorted()
                    .collect(Collectors.toList());
            return selectDataNodeFromList(dataNodeInfos, count, filename);
        }
    }

    /**
     * 从内存数据结构中移除DataNode的文件列表并返回
     *
     * @param hostname DataNode
     * @return 该DataNode的文件列表
     */
    public Map<String, FileInfo> removeFileByDataNode(String hostname) {
        replicaLock.writeLock().lock();
        try {
            //也移除了
            return filesByDataNode.remove(hostname);
        } finally {
            replicaLock.writeLock().unlock();
        }
    }


    public boolean dataNodeContainsFiles(String hostname, String filename) {
        replicaLock.readLock().lock();
        try {
            //todo NameNode 保存各个DataNode节点所保存的文件集合
            Map<String, FileInfo> files = filesByDataNode.getOrDefault(hostname, new HashMap<>(PrettyCodes.trimMapSize()));
            return files.containsKey(filename);
        } finally {
            replicaLock.readLock().unlock();
        }
    }

    /**
     * 为每一个 DataNode 实例节点 增加一个副本任务（基本是删除任务）
     *
     * @param fileInfo 文件信息
     */
    public void addReplica(FileInfo fileInfo) {
        replicaLock.writeLock().lock();
        try {
            //localhost:5601 a. png  DataNode_1
            //localhost:5602 a. png  DataNode_2
            // 获取该文件所属的 DataNode (由于只会保存一个)
            DataNodeInfo dataNode = dataNodes.get(fileInfo.getHostname());
            // 获取该文件对应的DataNode列表
            List<DataNodeInfo> dataNodeInfos = replicaByFilename.computeIfAbsent(fileInfo.getFileName(), k -> new ArrayList<>());
            // 文件目录树有可能被移动到.Trash目录下面了，所以除了正常查找一遍文件目录树之外，还要查一遍垃圾箱
            Node node = maybeInTrash(fileInfo.getFileName());

            if (node == null) {
                log.info("收到 DataNode 上报的存储信息，但是在 NameNode 内存目录树中不存在文件,下发命令让 DataNode 删除文件: [hostname={}, filename={}]", fileInfo.getHostname(), fileInfo.getFileName());
                RemoveReplicaTask task = new RemoveReplicaTask(fileInfo.getHostname(), fileInfo.getFileName());
                dataNode.addRemoveReplicaTask(task);
                return;
            }

            //文件副本数量 (配置文件中 默认两个)
            int replicaNum = Integer.parseInt(node.getAttr().getOrDefault(Constants.ATTR_REPLICA_NUM, String.valueOf(nameNodeConfig.getReplicaNum())));

            // 如果该文件的实际副本数量超过用户配置的数量，则让该 DataNode 删除文件
            if (dataNodeInfos.size() > replicaNum) {
                RemoveReplicaTask task = new RemoveReplicaTask(dataNode.getHostname(), fileInfo.getFileName());
                log.info("下发副本删除任务：[hostname={}, filename={}]", dataNode.getHostname(), fileInfo.getFileName());
                //为什么也不顺便删除这个呢？
                //dataNodeInfos.remove(0);
                dataNode.addRemoveReplicaTask(task);
                return;
            } else if (dataNodeInfos.size() < replicaNum) {
                // 副本数量没有超过，将文件信息维护起来
                //这里其实需要增加文件复制任务
                dataNodeInfos.add(dataNode);
            }
            //更新文件存储情况
            Map<String, FileInfo> files = filesByDataNode.computeIfAbsent(fileInfo.getHostname(), k -> new HashMap<>(PrettyCodes.trimMapSize()));
            //添加文件元属性之副本数量
            fileInfo.setReplicaNum(replicaNum);
            files.put(fileInfo.getFileName(), fileInfo);
            log.info("收到DataNode文件上报并更新filesByDataNode：[hostname={}, filename={}, {}]", fileInfo.getHostname(), fileInfo.getFileName(), filesByDataNode);
        } finally {
            replicaLock.writeLock().unlock();
        }
    }

    private Node maybeInTrash(String fileName) {
        Node node = diskNameSystem.listFiles(fileName);
        if (node != null) {
            return node;
        }
        String[] split = fileName.split("/");
        String[] newSplit = new String[split.length + 1];
        newSplit[0] = split[0];
        newSplit[1] = split[1];
        newSplit[2] = Constants.TRASH_DIR;
        System.arraycopy(split, 2, newSplit, 3, split.length - 2);
        String trashPath = String.join("/", newSplit);
        return diskNameSystem.listFiles(trashPath);
    }

    /**
     * 根据文件名选择一个可读的DataNode，并把不可读的DataNode从文件对应的DataNode数据结构中删除
     *
     * @param filename 文件名
     * @return 可读的DataNode
     */
    public DataNodeInfo chooseReadableDataNodeByFileName(String filename) {
        return chooseReadableDataNodeByFileName(filename, null);
    }

    /**
     * 根据文件名选择一个可读的DataNode，并把不可读的DataNode从文件对应的DataNode数据结构中删除
     *
     * @param filename         文件名
     * @param toRemoveDataNode 不可读的DataNode（比如已经宕机的节点）
     * @return 可读的DataNode
     */
    public DataNodeInfo chooseReadableDataNodeByFileName(String filename, DataNodeInfo toRemoveDataNode) {
        replicaLock.readLock().lock();
        try {
            List<DataNodeInfo> dataNodeInfos = replicaByFilename.get(filename);
            log.info("根据文件名选择一个可读的DataNode，并把不可读的DataNode从文件对应的DataNode数据结构中删除:{},{},{}", filename, replicaByFilename, toRemoveDataNode);
            if (dataNodeInfos == null) {
                return null;
            }
            if (toRemoveDataNode != null) {
                dataNodeInfos.remove(toRemoveDataNode);
            }
            if (dataNodeInfos.isEmpty()) {
                return null;
            }
            int size = dataNodeInfos.size();
            Random random = new Random();
            int i = random.nextInt(size);
            final DataNodeInfo dataNodeInfo = dataNodeInfos.get(i);
            log.info("根据文件名随机一个可读的 DataNode：{}", dataNodeInfo);
            return dataNodeInfo;
        } finally {
            replicaLock.readLock().unlock();
        }
    }

    /**
     * 根据文件名获取DataNode列表
     *
     * @param filename 文件名
     * @return DataNode信息
     */
    public List<DataNodeInfo> getDataNodeByFileName(String filename) {
        replicaLock.readLock().lock();
        try {
            return replicaByFilename.getOrDefault(filename, new ArrayList<>(PrettyCodes.trimListSize()));
        } finally {
            replicaLock.readLock().unlock();
        }
    }


    public DataNodeInfo getDataNode(String hostname) {
        return dataNodes.get(hostname);
    }

    public List<DataNodeInfo> getDataNodeInfoList() {
        return new ArrayList<>(dataNodes.values());
    }


    /**
     * 删除对应文件存储信息
     *
     * @param filename 文件名称
     * @return 被删除的文件信息
     */
    public FileInfo removeFileStorage(String filename, boolean delReplica) {
        replicaLock.writeLock().lock();
        try {
            List<DataNodeInfo> fileDataNodes = replicaByFilename.remove(filename);
            if (fileDataNodes == null) {
                return null;
            }
            FileInfo ret = null;
            for (Map<String, FileInfo> dataNodeFileInfo : filesByDataNode.values()) {
                FileInfo fileInfo = dataNodeFileInfo.remove(filename);
                if (fileInfo != null) {
                    ret = fileInfo;
                }
            }

            if (delReplica) {
                for (DataNodeInfo dataNodeInfo : fileDataNodes) {
                    DataNodeInfo dataNode = dataNodes.get(dataNodeInfo.getHostname());
                    RemoveReplicaTask task = new RemoveReplicaTask(dataNode.getHostname(), filename);
                    log.info("下发副本删除任务：[hostname={}, filename={}]", dataNode.getHostname(), filename);
                    dataNode.addRemoveReplicaTask(task);
                }
            }
            return ret;
        } finally {
            replicaLock.writeLock().unlock();
        }
    }

    public FileInfo getFileStorage(String filename) {
        replicaLock.readLock().lock();
        try {
            for (Map<String, FileInfo> map : filesByDataNode.values()) {
                FileInfo fileInfo = map.get(filename);
                if (fileInfo != null) {
                    return fileInfo;
                }
            }
            return null;
        } finally {
            replicaLock.readLock().unlock();
        }
    }

    /**
     * 等待直到DataNode上报文件
     *
     * @param filename 文件名
     */
    public void waitFileReceive(String filename, long timeout) throws NameNodeException, InterruptedException {
        long remainTimeout = timeout;
        //同步阻塞主
        synchronized (this) {
            //循环等待
            while (chooseReadableDataNodeByFileName(filename) == null) {
                if (remainTimeout < 0) {
                    throw new NameNodeException("等待文件上传确认超时：" + filename);
                }
                wait(10);
                remainTimeout -= 10;
            }
        }
    }

    /**
     * 为某个文件增加副本数量
     *
     * @param username 用户名
     * @param addNum   增加副本数量
     * @param filename 文件名
     */
    public void addReplicaNum(String username, int addNum, String filename) throws Exception {
        List<DataNodeInfo> dataNodeInfos = allocateDataNodes(username, addNum, filename);
        DataNodeInfo sourceDataNode = chooseReadableDataNodeByFileName(filename);
        Node node = diskNameSystem.unsafeListFiles(filename);
        int replicaNum = Integer.parseInt(node.getAttr().getOrDefault(Constants.ATTR_REPLICA_NUM,
                String.valueOf(nameNodeConfig.getReplicaNum())));
        node.getAttr().put(Constants.ATTR_REPLICA_NUM, String.valueOf(replicaNum + addNum));
        for (DataNodeInfo destDataNode : dataNodeInfos) {
            ReplicaTask task = new ReplicaTask(filename, sourceDataNode.getHostname(), sourceDataNode.getNioPort());
            log.info("创建副本复制任务：[filename={}, from={}, to={}]", filename, sourceDataNode.getHostname(), destDataNode.getHostname());
            destDataNode.addReplicaTask(task);
        }
    }

    /**
     * DataNode 是否存活的监控线程
     *
     * <pre>
     *     这里存在一种情况，假设一个DataNode宕机了，从DataNode集合中摘除
     * </pre>
     */
    private class DataNodeAliveMonitor implements Runnable {

        @Override
        public void run() {
            log.info(" DataNode 是否存活的监控线程已启动 ");
            //已经在 dataNodes 删除了
            Iterator<DataNodeInfo> iterator = dataNodes.values().iterator();
            List<DataNodeInfo> toRemoveDataNode = new ArrayList<>();
            //todo 3 nameNode后台检测心跳存活dateNode
            while (iterator.hasNext()) {
                DataNodeInfo next = iterator.next();
                long currentTimeMillis = System.currentTimeMillis();
                if (currentTimeMillis < next.getLatestHeartbeatTime()) {
                    continue;
                }
                log.info("DataNode存活检测超时，被移除：[hostname={}, current={}, nodeLatestHeartbeatTime={}]",
                        next, DateUtils.format(new Date(currentTimeMillis)), DateUtils.format(new Date(next.getLatestHeartbeatTime())));
                iterator.remove();
                //已经失活的 DataNode 节点
                toRemoveDataNode.add(next);
            }
            log.info("为已经失活的 DataNode 节点 创建副本复制任务 {}", toRemoveDataNode);
            //为每一个 失活的 节点 创建 副本复制任务
            for (DataNodeInfo info : toRemoveDataNode) {
                //重点逻辑 后面分析
                //根据宕机的节点来创建其保存数据的复制任务
                createLostReplicaTask(info);
            }
        }
    }
}
